package com.ruyuan.eshop.order.mq.listener;

import com.ruyuan.eshop.common.constants.RedisLockKeyConstants;
import com.ruyuan.eshop.common.exception.BaseBizException;
import com.ruyuan.eshop.common.message.PaidOrderSuccessMessage;
import com.ruyuan.eshop.common.redis.RedisLock;
import com.ruyuan.eshop.common.utils.JsonUtil;
import com.ruyuan.eshop.order.exception.OrderErrorCodeEnum;
import com.ruyuan.eshop.order.service.OrderFulFillService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author lxg
 * @Description:监听订单支付成功后的消息
 * @date 2022/4/622:33
 */
@Slf4j
@Component
public class PaidOrderSuccessListener implements MessageListenerConcurrently {
    @Autowired
    private RedisLock redisLock;
    @Autowired
    private OrderFulFillService orderFulFillService;
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try {
            for (MessageExt messageExt:list){
                String message=new String(messageExt.getBody());
                PaidOrderSuccessMessage paidOrderSuccessMessage= JsonUtil.json2Object(message, PaidOrderSuccessMessage.class);
                String orderId=paidOrderSuccessMessage.getOrderId();
                log.info("触发订单履约，orderId:{}", orderId);
                String key= RedisLockKeyConstants.ORDER_FULFILL_KEY+orderId;
                boolean lock = redisLock.lock(key);
                if (!lock){
                    log.error("order has not acquired lock，cannot fulfill, orderId={}",orderId);
                    throw new BaseBizException(OrderErrorCodeEnum.ORDER_FULFILL_ERROR);
                }
                try {
                    //2、触发订单履约逻辑
                    // 注意这里分布式锁加锁放在了本地事务外面
                    orderFulFillService.triggerOrderFulFill(orderId);
                }finally {
                    if (lock){
                        redisLock.unlock(key);
                    }
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
            log.error("consumer error", e);
            //本地业务执行失败触发重复消费
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
